package cn.hdax.test1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.List;

public class StuMr1 implements Tool {
    private Configuration configuration;

    public static void main(String[] args) throws Exception {
        int result = ToolRunner.run(new StuMr1(), args);
        System.exit(result);
    }


    @Override
    public int run(String[] args) throws Exception {
        System.setProperty("HADOOP_USER_NAME", "root");
        //声明作业
        Job job = Job.getInstance(this.getConf(), "stumr1");

        job.setJarByClass(StuMr1.class);

        // 取对业务有用的数据 info,age
        Scan scan = new Scan();
        scan.addColumn("info".getBytes(), "age".getBytes());
        TableMapReduceUtil.initTableMapperJob(
                "stu1".getBytes(), // 指定表名
                scan, // 指定扫描数据的条件
                StuMapper.class, // 指定mapper class
                Text.class,     // outputKeyClass mapper阶段的输出的key的类型
                IntWritable.class, // outputValueClass mapper阶段的输出的value的类型
                job, // job对象
                false
        );

        job.setReducerClass(StuReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        Path outputPath = new Path("/t176/out/age");
        FileSystem fs = FileSystem.get(this.getConf()) ;
        if(fs.exists(outputPath)) {
            fs.delete(outputPath,true);
        }

        FileOutputFormat.setOutputPath(job, outputPath);


        //提交作业
        return job.waitForCompletion(true) ? 0 : 1;
    }

    @Override
    public void setConf(Configuration conf) {
        this.configuration = new Configuration();
    }

    @Override
    public Configuration getConf() {
        return this.configuration;
    }
    public static class StuMapper extends TableMapper<Text, IntWritable> {
        Text outKey = new Text("age");
        IntWritable outValue = new IntWritable();

        // key是hbase中的行键
        // value是hbase中的所行键的所有数据
        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context)
                throws IOException, InterruptedException {
            boolean isContainsColumn = value.containsColumn("info".getBytes(), "age".getBytes());
            if (isContainsColumn) {
                List<Cell> listCells = value.getColumnCells("info".getBytes(), "age".getBytes());
                System.out.println("listCells:\t" + listCells);
                Cell cell = listCells.get(0);
                System.out.println("cells:\t" + cell);
                byte[] cloneValue = CellUtil.cloneValue(cell);
                String ageValue = Bytes.toString(cloneValue);
                outValue.set(Integer.parseInt(ageValue));
                context.write(outKey, outValue);
            }
        }
    }


    public static class StuReduce extends Reducer<Text, IntWritable, Text, DoubleWritable> {

        DoubleWritable outValue = new DoubleWritable();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values,Context context)
                throws IOException, InterruptedException {

            int count = 0;
            int sum = 0;
            for(IntWritable value : values) {
                count++;
                sum += value.get();
            }

            double avgAge = sum * 1.0 / count;
            outValue.set(avgAge);
            context.write(key, outValue);
        }

    }

}
